package com.atguigu.wuliu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.wuliu.app.func.BeanToJsonStrFunction;
import com.atguigu.wuliu.app.func.DimAsyncMapFunction;
import com.atguigu.wuliu.utils.DateFormatUtil;
import com.atguigu.wuliu.utils.DorisUtil;
import com.atguigu.wuliu.utils.KafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * TODO  物流域 : 当日各类卡车统计汇总
 *     当日各类卡车完成运输里程
 *     当日各类卡车完成运输次数
 *     当日各类卡车完成运输时长
 *     当日各类卡车平均每次运输时长
 *     当日各类卡车平均每次运输里程
 */
public class DwsTrandsTruckStatistics {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        env.enableCheckpointing(5000L);
//        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
//        checkpointConfig.setCheckpointStorage("hdfs://hadoop100:8020/wuliu/ck");
//        checkpointConfig.setCheckpointTimeout(60000L);
//        checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//        checkpointConfig.setMinPauseBetweenCheckpoints(2000L);
//        env.setStateBackend(new HashMapStateBackend());
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(3),Time.seconds(3000L)));

        //从dwd运输完成事实表中读取数据
        String topic="dwd_trans_trans_finish";
        String groupId="dws_trade_cargo_type_group";
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaSource(topic, groupId);
        DataStreamSource<String> ds = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");

        //将数据封装为json对象
        SingleOutputStreamOperator<JSONObject> jsonObjDs = ds.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String value, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                try {
                    JSONObject jsonObj = JSON.parseObject(value);
                    out.collect(jsonObj);
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            }
        });

        KeyedStream<JSONObject, String> idKeyedDs = jsonObjDs.keyBy(jsonObject -> jsonObject.getString("id"));

        //将json对象转换为实体类
        SingleOutputStreamOperator<DwsTrandsTruckStatisticsBean> beanDs = idKeyedDs.map(new MapFunction<JSONObject, DwsTrandsTruckStatisticsBean>() {
            @Override
            public DwsTrandsTruckStatisticsBean map(JSONObject jsonObj) throws Exception {
                //运输时长 = 实际到达时间 - 实际启动时间
                Long td = DateFormatUtil.toTs(jsonObj.getString("actual_end_time"), true) - DateFormatUtil.toTs(jsonObj.getString("actual_start_time"), true);
                DwsTrandsTruckStatisticsBean truckStatisticsBean = DwsTrandsTruckStatisticsBean.builder()
                        .truckId(jsonObj.getString("truck_id"))
                        .mileage(jsonObj.getInteger("actual_distance"))
                        .ct(1)
                        .td(td)
                        .ts(jsonObj.getLong("ts"))
                        .build();
                return truckStatisticsBean;
            }
        });

        //关联维度数据1
        SingleOutputStreamOperator<DwsTrandsTruckStatisticsBean> truckModelIdDs = AsyncDataStream.unorderedWait(
                beanDs,
                new DimAsyncMapFunction<DwsTrandsTruckStatisticsBean>("dim_truck_info") {
                    @Override
                    public void join(DwsTrandsTruckStatisticsBean obj, JSONObject jsonObj) {
                        obj.setTruckModelId(jsonObj.getString("truck_model_id"));
                    }

                    @Override
                    public String getKey(DwsTrandsTruckStatisticsBean obj) {
                        return obj.getTruckId();
                    }
                },
                60,
                TimeUnit.SECONDS
        );

        //关联维度数据2
        SingleOutputStreamOperator<DwsTrandsTruckStatisticsBean> withDimDs = AsyncDataStream.unorderedWait(
                truckModelIdDs,
                new DimAsyncMapFunction<DwsTrandsTruckStatisticsBean>("dim_truck_model") {
                    @Override
                    public void join(DwsTrandsTruckStatisticsBean obj, JSONObject jsonObj) {
                        obj.setTruckModelName(jsonObj.getString("model_name"));
                        obj.setTruckModelType(jsonObj.getString("model_type"));
                        obj.setTruckBrand(jsonObj.getString("brand"));
                    }

                    @Override
                    public String getKey(DwsTrandsTruckStatisticsBean obj) {
                        return obj.getTruckModelId();
                    }
                },
                60,
                TimeUnit.SECONDS
        );

        //设置水位线
        SingleOutputStreamOperator<DwsTrandsTruckStatisticsBean> withWatermarkDs = withDimDs.assignTimestampsAndWatermarks(
                WatermarkStrategy.<DwsTrandsTruckStatisticsBean>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner((e, t) -> e.getTs())
        );

        //按照卡车类别进行分组
        KeyedStream<DwsTrandsTruckStatisticsBean, String> truckModelIdKeyedDs = withWatermarkDs.keyBy(DwsTrandsTruckStatisticsBean::getTruckModelId);

        //设置窗口
        WindowedStream<DwsTrandsTruckStatisticsBean, String, TimeWindow> windowDs = truckModelIdKeyedDs.window(TumblingEventTimeWindows.of(Time.seconds(10L)));

        //聚合计算
        SingleOutputStreamOperator<DwsTrandsTruckStatisticsBean> reduceDs = windowDs.reduce(
                new ReduceFunction<DwsTrandsTruckStatisticsBean>() {
                    @Override
                    public DwsTrandsTruckStatisticsBean reduce(DwsTrandsTruckStatisticsBean v1, DwsTrandsTruckStatisticsBean v2) throws Exception {
                        v1.setMileage(v1.getMileage() + v2.getMileage()); //运输里程
                        v1.setCt(v1.getCt() + v2.getCt()); //运输次数
                        v1.setTd(v1.getTd() + v2.getTd()); //运输时长
                        return v1;
                    }
                },
                new WindowFunction<DwsTrandsTruckStatisticsBean, DwsTrandsTruckStatisticsBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<DwsTrandsTruckStatisticsBean> input, Collector<DwsTrandsTruckStatisticsBean> out) throws Exception {
                        String start = DateFormatUtil.toYmdHms(window.getStart());
                        String end = DateFormatUtil.toYmdHms(window.getEnd());
                        String curDate = DateFormatUtil.toPartitionDate(window.getStart());
                        DwsTrandsTruckStatisticsBean next = input.iterator().next();
                        next.setStt(start);
                        next.setEdt(end);
                        next.setCurDate(curDate);
                        out.collect(next);
                    }
                });

        reduceDs.print();

        //将数据写到doris中
        reduceDs.map(new BeanToJsonStrFunction<>())
                        .sinkTo(DorisUtil.getDorisSink("dws_trands_truck_statistics"));



        env.execute();
    }
}
